package com.levi.producer;

import com.levi.config.MessageConfirmCallback;
import com.levi.pojo.Orders;
import com.levi.util.JsonUtils;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;

/**
 * 发送消息 ，一会在test中进行测试
 * 使用AmqpTemplate的 convertAndSend()方法发送消息
 * ConfirmCallback
 */
@Component
public class TopicProducer implements RabbitTemplate.ConfirmCallback {

    public static final String EXCHANGE_NAME = "topic exchange";
//    public static final String QUEUT_NAME1 = "queue1";
//    public static final String QUEUT_NAME2 = "queue2";
    public static final String ROUTING_KEY1 = "dog.red";
    public static final String ROUTING_KEY2 = "dog.black";

    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * 发送普通消息
     */
    public void sendMsg1() {
        String msg = "This is mesage1,and routingkey is dog.red";
        rabbitTemplate.setConfirmCallback(new MessageConfirmCallback());
        //生产者实现ConfirmCallback接口的话可以这样写
//        rabbitTemplate.setConfirmCallback(this);
        //发送的消息为对象的话，要转为json字符串,例如下面
        Orders orders = new Orders(10001,2002,200);

        // 三个参数的方法发送消息之后会有消息确认
        rabbitTemplate.convertAndSend(EXCHANGE_NAME,ROUTING_KEY1, JsonUtils.objectToJson(orders),new CorrelationData(orders.getId().toString()));
        rabbitTemplate.convertAndSend(EXCHANGE_NAME,ROUTING_KEY1,msg);
        System.out.println("发送了消息: "+msg);
    }

    /**
     * 发送普通消息
     */
    public void sendMsg2() {
        String msg = "This is mesage2,and routingkey is dog.black";
        rabbitTemplate.convertAndSend(EXCHANGE_NAME,ROUTING_KEY2,msg);
        System.out.println("发送了消息: "+msg);
    }

    /**
     * 发送消息到带有ttl的队列中
     */
    public void sendTtl() {
        String msg = "发送消息到带ttl的队列中";
        rabbitTemplate.convertAndSend(EXCHANGE_NAME,"queue-ttl",msg);
        System.out.println("发送消息到带ttl的队列中: "+msg);
    }

    /**
     * 发送带有ttl的msg，给msg设置ttl
     */
    public void sendTtlMsg() {
        String msg = "带有ttl的msg";
        MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                //这里的时间是字符串
                message.getMessageProperties().setExpiration("20000");
                return message;
            }
        };
        rabbitTemplate.convertAndSend(EXCHANGE_NAME,"message-ttl",msg,messagePostProcessor);
    }

    /**
     * 解决消息幂等性问题
     * 给消息设置一个messageId  用order  id作为message id，实际生产中订单id一定是唯一的
     */
    public void sendMsgMideng() {
        Orders orders = new Orders(10001, 2001, 100);
        String jsonOrder = JsonUtils.objectToJson(orders);
        Message message = MessageBuilder.withBody(jsonOrder.getBytes(StandardCharsets.UTF_8)).setMessageId(orders.getId().toString()).build();
        rabbitTemplate.send(EXCHANGE_NAME,ROUTING_KEY1,message);
    }

    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        String orderId = correlationData.getId();
        System.out.println("orderId: "+orderId);
        //ack为false，交换机应答失败
        if (!ack) {
            //没有投递成功的消息写一个task定时检查本地消息冗余表，再次发送
            //或者再次投递消息
            sendMsg1();
            System.out.println("消息投递失败");
            return;
        }
        //ack为true，交换机应答成功，消息收到
        try {
            //修改消息冗余表的状态为收到消息   业务逻辑
            System.out.println("消息成功投递 >>>>> 本地消息表修改成功");
        }catch (Exception e){
            System.out.println("本地消息表修改失败，出现异常: "+e.getMessage());
        }
    }
}
